package com.dongdongshop.mq;

import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.pojo.TbOrder;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.service.UserScoreAccountService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Component
@RocketMQMessageListener(consumerGroup = "my_order_group",topic = "my_order_topic")
public class MQconsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @DubboReference
    private UserScoreAccountService userScoreAccountService;

    @Override
    public void onMessage(String message) {
        System.out.println("消息接收成功"+message);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                String topic = messageExt.getTopic();
                System.out.println("消息的主题:"+topic);
                String msgId = messageExt.getMsgId();
                System.out.println("消息的id:"+msgId);
                int reconsumeTimes = messageExt.getReconsumeTimes();
                System.out.println("消息的重试次数:"+reconsumeTimes);
                int queueId = messageExt.getQueueId();
                System.out.println("消息队列id:"+queueId);
                byte[] body = messageExt.getBody();
                try{
                    if (redisTemplate.hasKey(msgId)){
                        System.out.println("消息已经消费");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    String msg = new String(body,"utf-8");
                    TbOrder order = JSONObject.parseObject(msg, TbOrder.class);
                    String orderId = order.getOrderId().toString();
                    String price = order.getPayment().toString();
                    int i = price.indexOf(".");
                    String substring = price.substring(0, i);
                    TbUserScoreAccount userScoreAccount = new TbUserScoreAccount();
                        userScoreAccount.setUserId(orderId);
                        userScoreAccount.setScoreTotal(substring);
                        userScoreAccount.setScoreExpend(substring);
                        userScoreAccount.setStartDate(new Date());
                        userScoreAccount.setEntDate(new Date());
                        System.out.println("执行消费者业务逻辑,增加积分表");
                        userScoreAccountService.insertUserScoreAccount(userScoreAccount);
                    redisTemplate.boundValueOps(msgId).set("1");
                }catch (Exception e){
                    e.printStackTrace();
                    if (reconsumeTimes < 3){
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
